/**
 * 
 */
package experiment02;

import java.io.IOException;
import java.io.PrintStream;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;

class MyPathFilter implements PathFilter{
	String reg=null;
	public MyPathFilter(String reg) {
		this.reg=reg;
	}
	
	public boolean accept(Path path) {
		if(path.toString().matches(reg))
			return true;
		
		return false;
	}	
}

/**
 * 
 */
public class MergeFiles {
	
	
	private Configuration conf=null;
	private Path inputPath=null;
	private Path outputPath=null;
	
	public MergeFiles(String inputFolder,String outputFile) {
		this.inputPath=new Path(inputFolder);
		this.outputPath=new Path(outputFile);
		init();		
	}
	
	public MergeFiles() {
		init();
	}
	
	public void init() {
		conf=new Configuration();
		conf.set("fs.defaultFS", "hdfs://localhost:9000");
	}
	
	public void setInputPath(String input) {
		this.inputPath=new Path(input);
	}
	
	public void setOutputPath(String output) {
		this.inputPath=new Path(output);
	}
	
	public void Merge() throws IOException {
		FileSystem fsSourceFolder=FileSystem.get(URI.create(inputPath.toString()),conf);
		FileSystem fsDst=FileSystem.get(URI.create(outputPath.toString()),conf);
		
		FileStatus[] srcStatus=fsSourceFolder.listStatus(inputPath,new MyPathFilter(".*txt"));
		
		FSDataOutputStream fsdos=fsDst.create(outputPath);
		
		PrintStream ps=new PrintStream(System.out); 
		
		for(FileStatus sta:srcStatus) {
			System.out.println("\nPath :" + sta.getPath()+ "FileSize : "+ sta.getLen() + "Permission :"+ 
		sta.getPermission() + "Content :");
			FSDataInputStream fsdis=fsSourceFolder.open(sta.getPath());
			byte[] data=new byte[1024];
			int read=-1;
			while((read=fsdis.read(data))>0) {
				ps.write(data,0,read);
				fsdos.write(data,0,read);
			}
			fsdis.close();
		}
		ps.close();
		fsdos.close();
		fsDst.close();
		fsSourceFolder.close();		
	}

	/**
	 * @param args
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException {
		// TODO Auto-generated method stub
		
		String inputPath="hdfs://localhost:9000/user/abc/input";
		String outputFile="hdfs://localhost:9000/user/abc/output/merge.txt";
		
		MergeFiles merge=new MergeFiles(inputPath,outputFile);
		merge.Merge();
	}

}
